package io.simple.netty;

import injections.annotations.PostInject;
import injections.annotations.Property;
import io.simple.core.chain.Transport;
import org.jboss.netty.bootstrap.ServerBootstrap;
import org.jboss.netty.channel.ChannelPipeline;
import org.jboss.netty.channel.ChannelPipelineFactory;
import org.jboss.netty.channel.group.ChannelGroup;
import org.jboss.netty.channel.group.ChannelGroupFuture;
import org.jboss.netty.channel.group.DefaultChannelGroup;
import org.jboss.netty.channel.socket.ServerSocketChannel;
import org.jboss.netty.channel.socket.nio.NioServerSocketChannelFactory;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.net.InetSocketAddress;
import java.util.UUID;
import java.util.concurrent.Executors;

/**
 * Created with IntelliJ IDEA.
 * User: rehak
 * Date: 8/23/13
 * Time: 11:37 AM
 * To change this template use File | Settings | File Templates.
 */
public abstract class AbstractNettyTransport<IN, OUT> implements Transport<IN, OUT> {

    private static final Logger LOGGER = LoggerFactory.getLogger(AbstractNettyTransport.class);
    @Property
    private int[] ports = new int[0];
    @Property(optional = true)
    private String host = "0.0.0.0";
    @Property(optional = true)
    private boolean keepAlive = false;
    private ChannelGroup channelGroup = new DefaultChannelGroup(UUID.randomUUID().toString());
    private NioServerSocketChannelFactory factory;

    // inject constructor
    public AbstractNettyTransport() {

    }

    // public constructor
    public AbstractNettyTransport(int[] ports, String host, boolean keepAlive) {
        this.ports = ports;
        this.host = host;
        this.keepAlive = keepAlive;
        init();
    }

    @PostInject
    private void init() {
        // server socket channel factory
        this.factory = new NioServerSocketChannelFactory(Executors.newCachedThreadPool(), Executors.newCachedThreadPool()) {
            @Override
            public ServerSocketChannel newChannel(ChannelPipeline pipeline) {
                ServerSocketChannel serverSocketChannel = super.newChannel(pipeline);
                // add to channel group so we can close this whole thing clearly later on
                channelGroup.add(serverSocketChannel);
                return serverSocketChannel;
            }
        };
    }

    public abstract ChannelPipelineFactory getPipelineFactory();

    @Override
    public void start() throws Exception {
        // create bootstrap
        ServerBootstrap bootstrap = new ServerBootstrap(factory);
        // get pipeline factory
        ChannelPipelineFactory pipelineFactory = getPipelineFactory();
        // set factory
        bootstrap.setPipelineFactory(pipelineFactory);
        // tcp no delay
        bootstrap.setOption("child.tcpNoDelay", true);
        // tcp keep alive
        bootstrap.setOption("child.keepAlive", keepAlive);
        for (int port : ports) {
            InetSocketAddress socket = new InetSocketAddress(host, port);
            bootstrap.bind(socket);
            LOGGER.info("Server started at " + socket);
        }
    }

    @Override
    public void close() throws Exception {
        int size = channelGroup.size();
        try {
            LOGGER.info("Closing " + size + " channels");
            ChannelGroupFuture future = channelGroup.close();
            future.awaitUninterruptibly();
        } catch (Throwable t) {
            LOGGER.error("Error closing channels", t);
        }

        try {
            System.out.println("Releasing external resources...");
            factory.releaseExternalResources();
        } catch (Throwable t) {
            LOGGER.error("Error releasing external resources", t);
        }
    }

}
